package gox

import (
	"context"
	"errors"
	"fmt"
	"sync"

	"golang.org/x/sync/errgroup"
)

// ConsensusRunner 使用 errgroup 实现的并发共识运行器
type ConsensusRunner[T any] struct {
	// 并发执行的最大数量
	Concurrency int
	// 达到共识所需的最小计数
	MinConsensusCount int
	// 用于判断两个结果是否相同的比较函数
	CompareFunc func(a, b T) bool
}

// Task 代表一个需要执行的任务，返回泛型类型 T
type Task[T any] func(ctx context.Context) (T, error)

// RunUntilConsensus 并发执行tasks，直到某个结果达到共识，或所有任务执行完毕
func RunUntilConsensus[T any](ctx context.Context, tasks []Task[T], options ...func(*ConsensusRunner[T])) (T, error) {
	// 创建默认配置的运行器
	runner := &ConsensusRunner[T]{
		Concurrency:       5, // 默认并发数
		MinConsensusCount: 1, // 默认共识阈值
		CompareFunc: func(a, b T) bool {
			// 默认比较函数，对于简单类型有效
			return any(a) == any(b)
		},
	}

	// 应用选项
	for _, option := range options {
		option(runner)
	}

	// 零值，用于错误返回
	var zeroValue T

	if len(tasks) == 0 {
		return zeroValue, fmt.Errorf("no tasks provided")
	}

	// 验证并调整配置值
	if runner.Concurrency < 1 {
		runner.Concurrency = 1
	}
	if runner.Concurrency > len(tasks) {
		runner.Concurrency = len(tasks)
	}
	if runner.MinConsensusCount < 1 {
		runner.MinConsensusCount = 1
	}
	if runner.MinConsensusCount > len(tasks) {
		runner.MinConsensusCount = len(tasks)
	}

	// 创建可取消的上下文
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// 创建 errgroup 和自己的上下文
	g, gCtx := errgroup.WithContext(ctx)
	g.SetLimit(runner.Concurrency) // 设置并发限制

	// 用于收集结果的通道
	resultCh := make(chan T, len(tasks)) // 有缓冲的通道，避免阻塞

	// 用于收集任务错误的切片
	var taskErrors []error
	var taskErrorsMutex sync.Mutex
	var contextCanceled bool = false

	// 启动所有任务
	for i := range tasks {
		taskIndex := i // 捕获循环变量
		g.Go(func() error {
			// 执行任务
			result, err := tasks[taskIndex](gCtx)
			if err != nil {
				// 检查是否是上下文取消错误
				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
					taskErrorsMutex.Lock()
					contextCanceled = true
					taskErrorsMutex.Unlock()
					return nil // 不向 errgroup 报告错误
				}

				// 记录其他错误
				taskErrorsMutex.Lock()
				taskErrors = append(taskErrors, fmt.Errorf("task %d: %w", taskIndex, err))
				taskErrorsMutex.Unlock()
				return nil // 不中断其他任务
			}

			// 将结果发送到通道
			select {
			case resultCh <- result:
				return nil
			case <-gCtx.Done():
				return nil // 上下文取消，忽略结果
			}
		})
	}

	// 处理结果并达成共识的goroutine
	var results []T
	var counts []int
	var consensusReached bool
	var consensusResult T
	var maxCountResult T
	var maxCount int

	// 启动一个goroutine来收集结果
	done := make(chan struct{})
	go func() {
		defer close(done)

		tasksCompleted := 0
		for tasksCompleted < len(tasks) {
			select {
			case result, ok := <-resultCh:
				if !ok {
					return // 结果通道关闭
				}

				tasksCompleted++

				// 查找与当前结果相同的已有结果
				found := false
				for i, existingResult := range results {
					if runner.CompareFunc(result, existingResult) {
						counts[i]++
						found = true

						// 更新最大计数
						if counts[i] > maxCount {
							maxCount = counts[i]
							maxCountResult = existingResult
						}

						// 达到共识，立即取消其余任务
						if counts[i] >= runner.MinConsensusCount {
							consensusReached = true
							consensusResult = existingResult
							cancel() // 取消所有正在进行的任务
							return
						}
						break
					}
				}

				// 如果没有找到相同的结果，添加一个新条目
				if !found {
					results = append(results, result)
					counts = append(counts, 1)

					// 如果只需要一个结果就达到共识，立即返回
					if runner.MinConsensusCount == 1 {
						consensusReached = true
						consensusResult = result
						cancel() // 取消所有正在进行的任务
						return
					}

					// 更新最大计数
					if 1 >= maxCount {
						maxCount = 1
						maxCountResult = result
					}
				}

				// 检查是否仍有可能达到共识
				remainingTasks := len(tasks) - tasksCompleted
				canReachConsensus := false
				for _, count := range counts {
					if count+remainingTasks >= runner.MinConsensusCount {
						canReachConsensus = true
						break
					}
				}

				if !canReachConsensus && tasksCompleted < len(tasks) {
					// 已经无法达到共识，提前退出
					// 但是不要把这视为错误，我们仍然会返回最常见的结果
					cancel()
					return
				}

			case <-gCtx.Done():
				// 上下文取消，结束处理
				return
			}
		}
	}()

	// 等待所有任务完成
	waitErr := g.Wait()
	close(resultCh) // 关闭结果通道

	// 等待结果处理完成
	<-done

	// 如果达成共识，返回共识结果
	if consensusReached {
		return consensusResult, nil
	}

	// 如果没有达成共识但有结果，返回最常见的结果
	if len(results) > 0 {
		return maxCountResult, nil
	}

	// 检查是否是上下文取消导致的失败
	if contextCanceled || errors.Is(waitErr, context.Canceled) || errors.Is(waitErr, context.DeadlineExceeded) {
		return zeroValue, context.Canceled
	}

	// 如果所有任务都失败
	if len(taskErrors) > 0 {
		maxErrors := 5
		if len(taskErrors) > maxErrors {
			return zeroValue, fmt.Errorf("all tasks failed: %v and %d more errors",
				taskErrors[:maxErrors], len(taskErrors)-maxErrors)
		}
		return zeroValue, fmt.Errorf("all tasks failed: %v", taskErrors)
	}

	// 如果上下文被取消并且没有达成共识
	if waitErr != nil {
		return zeroValue, waitErr
	}

	return zeroValue, fmt.Errorf("no valid results returned")
}

// WithConcurrency 设置并发数
func WithConcurrency[T any](concurrency int) func(*ConsensusRunner[T]) {
	return func(runner *ConsensusRunner[T]) {
		runner.Concurrency = concurrency
	}
}

// WithMinConsensus 设置达到共识所需的最小计数
func WithMinConsensus[T any](count int) func(*ConsensusRunner[T]) {
	return func(runner *ConsensusRunner[T]) {
		runner.MinConsensusCount = count
	}
}

// WithCompareFunc 设置自定义比较函数
func WithCompareFunc[T any](compareFunc func(a, b T) bool) func(*ConsensusRunner[T]) {
	return func(runner *ConsensusRunner[T]) {
		runner.CompareFunc = compareFunc
	}
}
